package org.zjt.rabbit.demo.ack;

import org.zjt.rabbit.demo.common.ContantsConfig;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.Connection;
import org.springframework.amqp.rabbit.core.RabbitTemplate;


import java.io.IOException;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 消费者接受后setMandatory，通知发布者自己已经确认。消费者得到确认通知
 */
public class Consumer {

    public static void main(String[] args) throws IOException {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory(ContantsConfig.MQ_HOST);
        connectionFactory.setUsername(ContantsConfig.USER);
        connectionFactory.setPassword(ContantsConfig.PASSWORD);

        /**
         * 开启确认模式(消费者需要把publisherReturns设置为true)
         */
        connectionFactory.setPublisherReturns(true);


        Connection connection = connectionFactory.createConnection();


        //创建一个新的通道, 不支持事务, true表示支持
        final Channel channel = connection.createChannel(false);


        /**
         * 声明队列
         */
        channel.queueDeclare("hello-queue", true, true, false, null);


        /**
         * 将队列和交换器绑定
         */
        channel.queueBind("hello-queue", "hello-exchange", "hola");

/*

        RabbitTemplate template = new RabbitTemplate(connectionFactory);
        template.setQueue("hello-queue");

*/





        final AtomicInteger atomicInteger = new AtomicInteger(0);
        /**
         * 使用监听异步接收消息
         */
        final com.rabbitmq.client.Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String message = new String(body, "UTF-8");
                try {
                    System.out.println("Worker1 [x] Received '" + message + "'"+atomicInteger.get());
                    atomicInteger.incrementAndGet();

                    // basicAck参数：  消息标签 ，是否多个一起ack提交    是否重新发送到消息队列中requeue
                    if(atomicInteger.get() %2 ==0)
                        // 消息处理完成确认
                        channel.basicAck(envelope.getDeliveryTag(), false);
                    else
                        channel.basicNack(envelope.getDeliveryTag(), false, true);
                } finally {
                    System.out.println("Worker1 [x] Done");
                }
            }
        };
        // 消息消费完成确认
        channel.basicConsume("hello-queue", false, consumer);
    }
}
